package com.tc.vms.connection;

import com.tc.vms.VmsVersion;
import com.tc.vms.config.VmsRouterManager;
import com.tc.vms.consumer.DlqQueueMetaData;
import com.tc.vms.consumer.QueueMetaData;
import com.tc.vms.consumer.VmsConsumerProxy;
import com.tc.vms.exception.VmsClientErrorCode;
import com.tc.vms.exception.VmsClientException;
import com.tc.vms.iface.*;
import com.tc.vms.log.VmsClientLogger;
import com.tc.vms.metric.VmsMetricsManager;
import com.tc.vms.producer.ChannelMetaData;
import com.tc.vms.producer.VmsProducerProxy;
import org.slf4j.Logger;

import java.io.IOException;
import java.util.HashMap;
import java.util.Properties;
import java.util.concurrent.TimeUnit;

/**
 * Created by yonghua.zhang on 2015/12/11.
 */
public class VmsConnectionManager extends VmsServiceAdaptor {
    protected final static Logger LOGGER = VmsClientLogger.getLog();
    private final static int DEFAULT_RESOURCE_MAP_SIZE = 6;
    private final VmsRouterManager vmsRouterManager;
    private final HashMap<ChannelMetaData, VmsProducerProxy> chlConnections = new HashMap<ChannelMetaData, VmsProducerProxy>(DEFAULT_RESOURCE_MAP_SIZE);
    private final HashMap<QueueMetaData, VmsConsumerProxy> queueConnections = new HashMap<QueueMetaData, VmsConsumerProxy>(DEFAULT_RESOURCE_MAP_SIZE);
    private final VmsMetricsManager vmsMetricsManager = new VmsMetricsManager();
//    private final VmsHermesManager.VmsHermesReporter vmsHermesReporter = new VmsHermesManager.VmsHermesReporter();
//    private final VmsJmxReporter vmsJmxReporter = new VmsJmxReporter(vmsMetricsManager);
    //private final ScheduledExecutorService vmsManagerExecutor = Executors.newSingleThreadScheduledExecutor(new VmsThreadFactory("VmsConnectionManager-Scheduler"));

    public VmsConnectionManager(VmsRouterManager vmsRouterManager) {
        this.vmsRouterManager = vmsRouterManager;
    }

    public VmsRouterManager getVmsRouterManager() {
        return vmsRouterManager;
    }

    public VmsMetricsManager getVmsMetricsManager() {
        return vmsMetricsManager;
    }

    public synchronized <T extends ISubscriber> T createSubscriber(String queueName, String clientId, Properties props) throws VmsClientException {
        if (!isOpen) {
            throw new VmsClientException("VmsConnectionManager hasn't be init before new subscriber.", VmsClientErrorCode.VMS_SYS_ERR_FACTORY_NOT_INIT);
        }
        QueueMetaData queueMetaData = null;
        try {
            queueMetaData = vmsRouterManager.readQueueMetaData(queueName, clientId);
        } catch (IOException e) {
            LOGGER.error(String.format("readQueueMetaData the metadata of the queue[%s] error.", queueName), e);
            throw new VmsClientException("VmsRouterManager read queue metadata error.", VmsClientErrorCode.ZK_ERR_CODE_READ_QUEUE_EXCEPTION, e);
        }

        LOGGER.info("The metadata of the queue[{}] is [{}]", queueName, queueMetaData);

        if (!queueConnections.containsKey(queueMetaData)) {
            try {
                VmsConsumerProxy vmsConsumerProxy = new VmsConsumerProxy(this, queueMetaData, props);
                vmsConsumerProxy.init(null);
                queueConnections.put(queueMetaData, vmsConsumerProxy);
            } catch (Exception e) {
                LOGGER.error(String.format("Init vms consumer of the queue[%s] error.", queueName), e);
                throw new VmsClientException("create vms consumer error.", VmsClientErrorCode.VMS_SYS_ERR_CONSUMER_INIT_FAILED, e);
            }
        } else {
            LOGGER.info("Return the previous subscriber for the client instance is exists. {}", queueMetaData);
        }

        LOGGER.info("createSubscriber [{}] successfully, Vmsclient version is [{}]", queueName, VmsVersion.getDescByVer(VmsVersion.CurVersion));

        return (T) queueConnections.get(queueMetaData);

    }

    public synchronized <T extends ISubscriber> T createDLQSubscriber(String queueName, String clientId, Properties props) throws VmsClientException {
        if (!isOpen) {
            throw new VmsClientException("VmsConnectionManager hasn't be init before new subscriber.", VmsClientErrorCode.VMS_SYS_ERR_FACTORY_NOT_INIT);
        }
        DlqQueueMetaData dlqQueueMetaData = null;
        try {
            dlqQueueMetaData = vmsRouterManager.readDLQQueueMetaData(queueName, clientId);
        } catch (IOException e) {
            LOGGER.error("readQueueMetaData the metadata of the dlq_queue[{}] error.", queueName, e);
            throw new VmsClientException("VmsRouterManager read dlq_queue metadata error.", VmsClientErrorCode.ZK_ERR_CODE_READ_QUEUE_EXCEPTION);
        }
        LOGGER.info("The metadata of the dlq_queue[{}] is [{}]", queueName, dlqQueueMetaData);
        if (!queueConnections.containsKey(dlqQueueMetaData)) {
            try {
                VmsConsumerProxy vmsConsumerProxy = new VmsConsumerProxy(this, dlqQueueMetaData, props);
                vmsConsumerProxy.init(null);
                queueConnections.put(dlqQueueMetaData, vmsConsumerProxy);
            } catch (VmsClientException e) {
                LOGGER.error("Init vms consumer of the dlq_queue[{}] error.", queueName, e);
                throw new VmsClientException(e.getErrMsg(), VmsClientErrorCode.VMS_SYS_ERR_CONSUMER_INIT_FAILED);
            }
        } else {
            LOGGER.info("Return the previous subscriber for the client instance is exists. {}", dlqQueueMetaData);
        }

        LOGGER.info("createSubscriber [{}] successfully, Vmsclient version is [{}]", queueName, VmsVersion.getDescByVer(VmsVersion.CurVersion));

        return (T) queueConnections.get(dlqQueueMetaData);
    }

    public synchronized <T extends IPublisher> T createPublisher(String channelName, String clientId, Properties props) throws VmsClientException {
        if (!isOpen) {
            throw new VmsClientException("VmsConnectionManager hasn't be init before create publisher.", VmsClientErrorCode.VMS_SYS_ERR_FACTORY_NOT_INIT);
        }

        ChannelMetaData channelMetaData = null;
        try {
            channelMetaData = vmsRouterManager.readChannelMetaData(channelName, clientId);
        } catch (IOException e) {
            LOGGER.error("readChannelMetaData the metadata of the channel [{}] error.", channelName, e);
            throw new VmsClientException(String.format("VmsRouterManager read Channel of [%s] metadata error.",
                    channelName), VmsClientErrorCode.ZK_ERR_CODE_READ_CHANNEL_EXCEPTION, e);
        }

        LOGGER.info("The metadata of the channel[{}] is [{}]", channelName, channelMetaData);

        if (!chlConnections.containsKey(channelMetaData)) {
            try {
                chlConnections.put(channelMetaData, new VmsProducerProxy(this, channelMetaData, props));
            } catch (VmsClientException e) {
                LOGGER.error("Init vms producer of the channel [{}] error.", channelName, e);
                throw new VmsClientException("create vms producer error.", VmsClientErrorCode.VMS_SYS_ERR_CANNOT_NEWPRODUCER, e);
            }
        } else {
            LOGGER.info("Return previouse publisher for Instance user are exists. {}", channelMetaData);
        }
        LOGGER.info("createPublisher [{}] successfully, Vmsclient version is [{}]", channelName, VmsVersion.getDescByVer(VmsVersion.CurVersion));
        return (T) chlConnections.get(channelMetaData);
    }

    public synchronized void removeConnection(VmsConnectionWrap con) {
        //TODO: Normally the function couldn't be invoked
        LOGGER.warn("Don't support remove connection.");
    }

    public synchronized void updateConnectionWrap(VmsConnectionWrap vmsConnection, VmsMetaData oldMetaData, VmsMetaData newMetaData) {
        if (vmsConnection.getVmsMetaData() == oldMetaData) {
            LOGGER.info("updateConnectionWrap remove oldMetaData and add newMetaData to VmsConnectionManager pool.");
            if (oldMetaData instanceof ChannelMetaData) {
                VmsProducerProxy vmsProducerProxy = chlConnections.remove(oldMetaData);
                vmsProducerProxy.setVmsMetaData(newMetaData);
                chlConnections.put((ChannelMetaData) newMetaData, vmsProducerProxy);
            } else if (oldMetaData instanceof QueueMetaData) {
                VmsConsumerProxy vmsConsumerProxy = queueConnections.remove(oldMetaData);
                vmsConsumerProxy.setVmsMetaData(newMetaData);
                queueConnections.put((QueueMetaData) newMetaData, vmsConsumerProxy);
            }
        }
    }

    @Override
    public synchronized boolean init(IContext context) throws VmsClientException {
        isOpen = true;
        try {
//            vmsHermesReporter.reportBootParams();
//            vmsJmxReporter.start(5, TimeUnit.MINUTES);
            LOGGER.info("VmsConnectionManager initialize successfully.");
        } catch (Exception e) {
        }
        return isOpen;
    }

    @Override
    public synchronized void shutdown() {
        if (isOpen) {
            for (VmsConnectionWrap vmsConnectionWrap : chlConnections.values()) {
                if (vmsConnectionWrap != null && vmsConnectionWrap.isOpen()) {
                    vmsConnectionWrap.shutdown();
                }
            }
            chlConnections.clear();

            for (VmsConnectionWrap vmsConnectionWrap : queueConnections.values()) {
                if (vmsConnectionWrap != null && vmsConnectionWrap.isOpen()) {
                    vmsConnectionWrap.shutdown();
                }
            }
            queueConnections.clear();
//            vmsJmxReporter.shutdown();
            LOGGER.info("VmsConnectionManager shutdown successfully.");
        }
    }
}
